package com.vhall.component.service.room.config;

import com.vhall.component.service.room.listener.BigDataBroadcastRecordReportStreamListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.time.Duration;

/**
 * 大数据上报
 *
 * @author yuanzh
 * @date 2021/11/26
 */
@Slf4j
@Configuration
public class BigDataReportStreamConfig {

//	@Value("${spring.redis.host}")
//	private String host;
//	@Value("${spring.redis.port}")
//	private Integer port;
//	@Value("${spring.redis.database}")
//	private Integer database;
//	@Value("${spring.redis.password}")
//	private String password;

    @Value("${vhall.bigdata.broadcast-record-name:bigdata-broadcast-record-stream}")
    private String broadcastRecordReportStreamName;

    @Value("${vhall.bigdata.broadcast-record-group-name:bigdata-broadcast-record-group}")
    private String broadcastRecordReportGroupName;

    @Value("${vhall.bigdata.broadcast-record-consumer-name:bigdata-broadcast-record-consumer}")
    private String broadcastRecordReportConsumerName;

    /**
     * 初始化角色数据
     */
//	@PostConstruct
//	public void initStream() {
//		RedisClient redisClient = null;
//		try {
//			log.warn("========> 房间组件 stream 初始化 start");
//			redisClient = RedisUtil.getRedisClient(host, port, password, database);
//
//			StatefulRedisConnection<String, String> connect = redisClient.connect();
//			// 初始化大数据-房间开播记录stream
//			String result = RedisUtil.initStream(connect, broadcastRecordReportStreamName, broadcastRecordReportGroupName);
//			log.warn("========> 房间组件 stream 初始化结束，stream:[{}]，group:[{}]，result:[{}]", broadcastRecordReportStreamName,
//					 broadcastRecordReportGroupName, result);
//
//		} catch (Exception ex) {
//			log.error("========> 房间组件 Bigdata Broadcast Record Stream 初始化异常", ex);
//		} finally {
//			if (null != redisClient) {
//				redisClient.connect().close();
//			}
//		}
//	}
    @Bean(name = "BigDataBroadcastRecordReport-StreamMessageListener")
    public StreamMessageListenerContainer<String, ObjectRecord<String, String>> getStreamMessageListenerContainer(
            RedisConnectionFactory factory, ThreadPoolTaskExecutor threadPoolTaskExecutor, BigDataBroadcastRecordReportStreamListener broadcastRecordReportStreamListener) {

        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder().pollTimeout(Duration.ofSeconds(1)).batchSize(10).targetType(String.class)
                        .executor(threadPoolTaskExecutor).build();

        StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer
                .create(factory, options);

        // 注册消费者对象
        // 导出相关
        container.register(buildReadRequest(broadcastRecordReportStreamName, broadcastRecordReportGroupName,
                broadcastRecordReportConsumerName), broadcastRecordReportStreamListener);

        container.start();
        return container;
    }

    /**
     * 构建readRequest
     *
     * @param streamName   stream名
     * @param groupName    组名
     * @param consumerName 消费者名
     * @return readRequest
     */
    private StreamMessageListenerContainer.StreamReadRequest<String> buildReadRequest(String streamName, String groupName,
                                                                                      String consumerName) {
        //指定消费最新的消息
        StreamOffset<String> streamOffset = StreamOffset.create(streamName, ReadOffset.lastConsumed());
        //创建消费者
        Consumer consumer = Consumer.from(groupName, consumerName);
        return StreamMessageListenerContainer.StreamReadRequest.builder(streamOffset)
                .errorHandler(error -> log.error(error.getMessage()))
                .cancelOnError(e -> false).consumer(consumer)
                //关闭自动ack确认
                .autoAcknowledge(false).build();
    }
}
